package org.ala.tree.lsm.wal;

import java.io.IOException;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.atomic.AtomicInteger;

import org.ala.tree.lsm.tools.FileTools;

/**
 * wal写线程
 *
 * @author ala
 * @date 2025年5月26日
 */
public class WalWriterThread extends Thread {
	/**
	 * 	文件id，切片id
	 */
	protected String id;
	protected AtomicInteger shardId;
	/**
	 * 	文件目录，当前文件名
	 */
	protected String dir, fileName;
	/**
	 * 	文件容量，当前已写总容量，缓冲区容量，当前已写缓冲区容量
	 */
	protected long capacity, write, bufferSize, bufferWrite;
	/**
	 * 	文件句柄通道，当前mmap映射
	 */
	protected FileChannel fileChannel;
	protected MappedByteBuffer buffer;
	public WalWriterThread(Runnable r, String rootDir, String walDir, String id, AtomicInteger shardId, long capacity, long bufferSize) throws IOException {
		super(r);
		
		this.shardId = shardId;
		this.capacity = capacity;
		this.bufferSize = bufferSize;
		
		dir = String.format("%s/%s/%s", rootDir, walDir, id);
		FileTools.tryMakeDir(dir);
		
		fileName = String.format("%d.wal", shardId.getAndIncrement());
		this.setName(String.format("WAL_THREAD_%s", fileName));
		this.setDaemon(false);
		
		fileChannel = FileChannel.open(Path.of(String.format("%s/%s", dir, fileName)), StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.CREATE);
		write = fileChannel.size();
		buffer = fileChannel.map(MapMode.READ_WRITE, write, bufferSize);
	}
	
	
	/**
	 * 	关闭
	 * @throws IOException 
	 */
	public void close() throws IOException {
		try {
			buffer.force();
		} finally {
			fileChannel.close();
			buffer = null;
			fileChannel = null;
		}
		
	}
	/**
	 * 	写文件
	 * @throws IOException 
	 */
	public void write(byte type, byte[] key, byte[] val) throws IOException {
		int need = Wal.HEAD_LEN + key.length + val.length;
		check(need);
		
		//	写头（5字节）
		buffer.put(type);
		buffer.putInt(key.length);
		buffer.putInt(val.length);
		//	写体
		buffer.put(key);
		buffer.put(val);
		
		//	更新指针
		write += need;
		bufferWrite += need;
	}
	/**
	 * 	检测文件是否够写，不够写新开一个
	 * 	@throws IOException 
	 */
	protected boolean checkFile(int need) throws IOException {
		if (write + need >= capacity) {
			try {
				if (fileChannel != null) {
					buffer.force();
					fileChannel.close();
				}
			} finally {
				fileName = String.format("%d.wal", shardId.getAndIncrement());
				fileChannel = FileChannel.open(Path.of(String.format("%s/%s", dir, fileName)), StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.CREATE);
				
				write = fileChannel.size();
				buffer = fileChannel.map(MapMode.READ_WRITE, write, bufferSize);
				bufferWrite = 0;
			}
			return true;
		}
		return false;
	}
	/**
	 * 	检测当前剩余缓冲容量是否够写，不够新开一个buffer
	 * 	@throws IOException 
	 */
	protected boolean checkBuffer(int need) throws IOException {
		if (bufferWrite + need >= bufferSize) {
			//	特殊的，如果bufferSize本来就不够。bufferSize改为 > need的2的正次幂
			if (bufferWrite == 0) {
				bufferSize = 1 << (32 - Integer.numberOfLeadingZeros(need));
			}
			
			try {
				if (buffer != null) {
					buffer.force();
				}
			} finally {
				buffer = fileChannel.map(MapMode.READ_WRITE, write, bufferSize);
				bufferWrite = 0;
			}
			return true;
		}
		return false;
	}
	/**
	 * 	检测容量
	 * @throws IOException 
	 */
	protected void check(int need) throws IOException {
		if (!checkFile(need)) {
			checkBuffer(need);
		}
	}
	
}
